package rpc.frame.client.request;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import rpc.frame.core.cache.ServiceProviderCache;
import rpc.frame.core.data.RpcRequest;
import rpc.frame.core.data.RpcResponse;
import rpc.frame.core.netty.codec.FrameDecoder;
import rpc.frame.core.netty.codec.FrameEncoder;
import rpc.frame.core.netty.codec.RpcRequestEncoder;
import rpc.frame.core.netty.codec.RpcResponseDecoder;
import rpc.frame.core.netty.handler.RpcResponseHandler;
import rpc.frame.core.provider.ServiceProvider;
import rpc.frame.core.request.ChannelMapping;
import rpc.frame.core.request.RequestPromise;
import rpc.frame.core.request.RpcRequestHolder;

import javax.annotation.Resource;
import java.util.List;

@Component
@Slf4j
public class RpcRequestManager {

    @Resource
    private ServiceProviderCache cache;


    /**
     * 具体u实现远程服务调用
     * 1.获取当前请求发给谁->从请求中 获取提供者地址
     * 2.根据提供者的ip 和端口 进行rpc 调用
     * @param request
     * @return
     */
    public RpcResponse sendRequest(RpcRequest request) {
        //1.根据类名 获取提供者
        List<ServiceProvider> serviceProviders = cache.get(request.getClassName());
        if(CollectionUtils.isEmpty(serviceProviders)){
            log.error("接口{} 没有找到相关的服务提供者",request.getClassName());
            throw  new RuntimeException(request.getClassName()+"：获取服务提供者失败");
        }
        //获取第一个提供者 (后续再负载均衡->现在有点憔悴)
        ServiceProvider provider=serviceProviders.get(0);

        return requestNetty(provider,request);
    }

    /**
     * 1.连接服务提供者
     * 2.通信
     * 3.加密解密
     * 4.参数返回
     * @param provider
     * @param request
     * @return
     */
    private RpcResponse requestNetty(ServiceProvider provider,RpcRequest request){
        Channel channel;
        //如果没有建立过channel 的话 建立、已经建立过的直接从缓存中获取
        if(!RpcRequestHolder.channelExist(provider.getServerIp(),provider.getRpcPort())){
            NioEventLoopGroup eventExecutors = new NioEventLoopGroup(0, new DefaultThreadFactory("frame-client"));
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventExecutors)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            //encoder
                            pipeline.addLast("firstEncode",new FrameEncoder());
                            pipeline.addLast("secondEncode",new RpcRequestEncoder());//对response 解码

                            //decoder
                            pipeline.addLast("firstDecode",new FrameDecoder());
                            pipeline.addLast("secondDecode",new RpcResponseDecoder());//对request 加码
                            pipeline.addLast("rpcResponse",new RpcResponseHandler());
                        }
                    });

            try {
                ChannelFuture connect = bootstrap.connect(provider.getServerIp(), provider.getRpcPort()).sync();
                if (connect.isSuccess()){
                    RpcRequestHolder.addChannel(new ChannelMapping(
                            provider.getServerIp()
                            ,provider.getRpcPort()
                            , connect.channel()
                    ));
                }
            } catch (Exception e) {
                log.error("建立连接失败，msg={}",e);
                throw new RuntimeException(e);
            }
        }

        try {
            //从缓存中获取 channel
            channel=RpcRequestHolder.getChannel(provider.getServerIp(),provider.getRpcPort());
            //建立 promise
            RequestPromise requestPromise = new RequestPromise(channel.eventLoop());
            RpcRequestHolder.put(request.getRequestId(),requestPromise);
            requestPromise.addListener(future -> {
                RpcRequestHolder.remove(request.getRequestId());
            });

            //发送消息
            channel.writeAndFlush(request);

            RpcResponse response = (RpcResponse) requestPromise.get();
            return response;
        } catch (Exception e) {
            log.error("网络通讯错误：",e);
            RpcResponse response = RpcResponse.builder().requestId(request.getRequestId()).cause(e).build();
            return response;
        }
    }
}
